package com.nami.logic.disruptor;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;
import com.nami.logic.disruptor.event.TaskEvent;
import com.nami.logic.disruptor.factory.TaskEventFactory;
import com.nami.logic.disruptor.handler.TaskEventHandler;
import com.nami.store.api.service.RemoteConverTaskService;
import com.nami.store.api.service.RemoteL2CacheService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;

@Slf4j
public class TaskDisruptor {

    private final Disruptor<TaskEvent> disruptor;
    private final RingBuffer<TaskEvent> ringBuffer;

    public TaskDisruptor(RemoteConverTaskService taskService, RemoteL2CacheService l2CacheService, RedissonClient redissonClient) {
        // 初始化 Disruptor
        int bufferSize = 1024; // 环形缓冲区大小
        this.disruptor = new Disruptor<>(
            new TaskEventFactory(),
            bufferSize,
            DaemonThreadFactory.INSTANCE
        );

        // 设置事件处理器
        this.disruptor.handleEventsWith(new TaskEventHandler(taskService, l2CacheService, redissonClient));

        // 启动 Disruptor
        this.disruptor.start();
        this.ringBuffer = disruptor.getRingBuffer();
        log.info("[disruptor] 初始化内存mq success. ");
    }

    // 添加任务到 Disruptor
    public void add(String taskKey) {
        long sequence = ringBuffer.next(); // 获取下一个可用的序列号
        try {
            TaskEvent event = ringBuffer.get(sequence); // 获取对应的事件对象
            event.setTaskKey(taskKey); // 设置任务数据
        } catch (Exception var3) {
            log.error("[disruptor] error:", var3);
        } finally {
            ringBuffer.publish(sequence); // 发布事件
        }
    }

    // 关闭 Disruptor
    public void shutdown() {
        disruptor.shutdown();
    }
}
